#include "server.h"

extern int worker_add_msg(int worker_idx, char *request);

static struct
{/*{{{*/
    unsigned long thrd_id;
    EP ep;
    EP_SLOT register_slot;
    EP_SLOT server_slot;
    char *msg;
}/*}}}*/
listener_ctl;

static int _register(SOCKET new_sock)
{/*{{{*/
#define _TIMEOUT            2
    char *msg=listener_ctl.msg;
    int msg_size;
    char *rtn;
    int i;
    int res;

    res=sock_read(new_sock, (unsigned char *)msg, &msg_size,
            _TIMEOUT, WRAP);
    if(res)
    {
        errlog("%s: sock_read register slot fail", __func__);
        sock_close(new_sock);
        return(-1);
    }

    char para_desc[100];
    PARA para;
    char **svcs=NULL;
    sprintf(para_desc,
            "svrname(string:1-%d)"
            "svcs(array) { name(string:1-%d) }"
            "pid(number:1-8)",
            NAME_SIZE, NAME_SIZE);
    res=para_load(msg, para_desc, &para);
    if(res)
    {
        if(res > 0)
            errlog("%s: para_load msg(%.20s) no.%d error, impossible",
                    __func__, msg, res);
        else
            errlog("%s: para_load msg(%.20s) fail, res:%d, impossible",
                    __func__, msg, res);
        rtn=SOAR_ERR_PARA;
        goto GO_OUT;
    }
    char *svrname;
    para_ref_by_name(para, "svrname", &svrname);
    char *p;
    para_ref_by_name(para, "svcs", &p);
    int svc_num=atoi(p);
    ALLOC_MULTI_DOERR(svcs, char *, svc_num,
            errlog("%s: alloc svcs fail", __func__);
            rtn=SOAR_ERR_RESOURCE;
            goto GO_OUT);
    for(i=0; i<svc_num; ++i)
    {
        para_ref_by_index(para, i+2, NULL, &svcs[i]);
    }
    para_ref_by_name(para, "pid", &p);
    pid_t pid=atoi(p);

    pthread_mutex_lock(&stat_info._svr_.lock);   // lock

    // 找到svr
    int idle_idx=-1;
    for(i=0; i<SVR_MAX_NUM; ++i)
    {
        struct _svr_info *svr=&stat_info._svr_.svrs[i];
        if(svr->pid > 0)
        {
            if(strcmp(svr->name, svrname) == 0)
            {
                errlog("%s: svr_%s duplidate, must downsvr first",
                        __func__, svrname);
                rtn=SOAR_ERR_SVR_DUPLICATE;
                goto UNLOCK;
            }
        }
        else
        {
            if(idle_idx == -1) idle_idx=i;
        }
    }
    if(idle_idx == -1)
    {
        errlog("%s: svr full, BAD", __func__);
        rtn=SOAR_ERR_SVR_FULL;
        goto UNLOCK;
    }
    struct _svr_info *svr=&stat_info._svr_.svrs[idle_idx];
    strcpy(svr->name, svrname);

    // 检查是否与其它svr的svc重名
    for(i=0; i<svc_num; ++i)
    {
        struct rb_link *link=rb_find(&stat_info._svr_.svcs, svcs[i]);
        if(link)
        {
            errlog("%s: svr_%s.svc(%s) duplicate",
                    __func__, svr, svcs[i]);
            rtn=SOAR_ERR_SVC_DUPLICATE;
            goto UNLOCK;
        }
    }
    // 加入svc
    for(i=0; i<svc_num; ++i)
    {
        struct _svc_rec *svc;
        ALLOC_DOERR(svc, struct _svc_rec,
                errlog("%s: alloc svc_rec fail", __func__);
                rtn=SOAR_ERR_RESOURCE;
                goto UNLOCK;
                );
        strcpy(svc->name, svcs[i]);
        svc->svr_idx=idle_idx;
        svc->call=0;
        tree_setkey_string(svc, name, link_svc);
        listp_init(&svc->link_svr);
        res=rb_insert(&stat_info._svr_.svcs, &svc->link_svc);
        if(res)
        {
            errlog("%s: insert svc(%s) into stat_info._svr_.svcs fail, "
                    "res:%d, BAD",
                    __func__, svcs[i], res);
            free(svc);
        }
        listp_add_tail(&svc->link_svr, &svr->svcs);
#ifdef _MY_DEBUG
        errlog("svc: %s", svcs[i]);
#endif
    }
    svr->pid=pid;

    errlog("%s: svr_%s register", __func__, svr);
    rtn=SOAR_ERR_OK;

UNLOCK:
    pthread_mutex_unlock(&stat_info._svr_.lock);

GO_OUT:
    FREE(svcs);
    para_free(para);

    CLR_BUF(msg);
    para_put_item(msg, rtn);
    msg_size=strlen(msg);
    res=sock_write(new_sock, (unsigned char *)msg, &msg_size,
            _TIMEOUT, WRAP);
    if(res)
    {
        errlog("%s: sock_write register slot fail", __func__);
    }
    sock_close(new_sock);

    return(0);
}/*}}}*/

static int _newconn(SOCKET new_sock)
{/*{{{*/
#define _MAX_CONN_NUM        100
    int idle_num=0;
    int idle[cnfg_info.threads];
    int min_busy_conn_num=_MAX_CONN_NUM;
    int min_busy_idx=-1;
    int worker_idx;

    pthread_mutex_lock(&stat_info._worker_.lock);

    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        if(worker->conn_num == 0)
        {   idle[idle_num++]=worker_idx;   }
        else
        {
            if(min_busy_conn_num > worker->conn_num)
            {
                min_busy_conn_num=worker->conn_num;
                min_busy_idx=worker_idx;
            }
        }
    }
    if(idle_num > 0) // 选取无连接的worker
    {
        worker_idx=idle[random_int(0, idle_num-1, NULL)];
#ifdef _MY_DEBUG
        errlog("%s: choose idle thread(%d)",
                __func__, worker_idx);
#endif
    }
    else if(min_busy_conn_num < _MAX_CONN_NUM) // 选取连接最少的worker
    {
        worker_idx=min_busy_idx;
    }
    else
    {
        errlog("%s: worker full, BAD", __func__);
        pthread_mutex_unlock(&stat_info._worker_.lock);
        sock_close(new_sock);
        return(-1);
    }
    ++stat_info._worker_.worker[worker_idx].conn_num;
#ifdef _MY_DEBUG
    errlog("%s: worker(%d) conn_num:%d", __func__, worker_idx,
            stat_info._worker_.worker[worker_idx].conn_num);
#endif

    pthread_mutex_unlock(&stat_info._worker_.lock);

    // 发送到worker
    char *request=listener_ctl.msg;
    CLR_BUF(request);
    para_put_item(request, "NEWCONN");
    para_put_item_int(request, new_sock);
    para_put_item_int64(request, (int64_t)curr_timestamp(NULL));
    int res=worker_add_msg(worker_idx, request);
    if(res)
    {
        errlog("%s: listener send msg to worker(%d) fail, res:%d, BAD",
                __func__, worker_idx, res);
        sock_close(new_sock);
        return(-1);
    }

    return(0);
}/*}}}*/

static int _conn(struct ep_active_slot *active_slot)
{/*{{{*/
    SOCKET new_sock=active_slot->_u._conn.sock;

    if(active_slot->slot == listener_ctl.register_slot)
    {
        _register(new_sock);
    }
    else
    {
        _newconn(new_sock);
    }

    return(0);
}/*}}}*/

static void _free_svcs(struct _svr_info *svr)
{/*{{{*/
    struct _svc_rec *svc;

    while(!listp_is_empty(&svr->svcs))
    {
        svc=listp_entry(svr->svcs.next, struct _svc_rec,
                link_svr);
        if(!rb_delete_link(&stat_info._svr_.svcs, &svc->link_svc))
        {
            errlog("%s: delete svc(%s) from stat_info._svr_.svcs fail, "
                    "impossible",
                    __func__, svc->name);
        }
        listp_del(&svc->link_svr);
        free(svc);
    }
    CLR_BUF(svr->name);
    svr->pid=-1;
}/*}}}*/

static void _listener()
{/*{{{*/
    unsigned int quit=IS_FALSE;
    struct ep_active_slot *active_list;
    time_t last_operate=0;
    time_t last_stat=0;
    time_t now;
    int res;
    
    while(!quit)
    {
        res=ep_poll(listener_ctl.ep, &active_list);
        if(res)
        {
            errlog("%s: ep_poll fail, res:%d, impossible",
                    __func__, res);
            continue;
        }
        while(active_list)
        {/*{{{*/
            switch(active_list->type)
            {
                case EP_SLOT_CONN:
                    _conn(active_list);
                    break;
                case EP_SLOT_ABORT:
                    errlog("%s: revoke slot(sock:%d,idx:%d) error(%d), "
                            "impossible",
                            __func__, active_list->_u._abort.sock,
                            active_list->slot_idx,
                            active_list->_u._abort.error);
                    break;
                default:
                    errlog("%s: listener poll type(%s) invalid, "
                            "impossible",
                            __func__, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }/*}}}*/

        now=curr_timestamp(NULL);

        // check shutdown
#define CHECK_OPERATE_INFO_INTERVAL         1
        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {/*{{{*/
            pthread_mutex_lock(&operate_info._listener_.lock); //lock
            quit=operate_info._listener_.quit;
            if(quit)
            {
                operate_info._listener_.state=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._listener_.lock); //unlock

            last_operate=now;
        }/*}}}*/

        // check svr
#define CHECK_STAT_INFO_INTERVAL            3
        if(now - last_stat > CHECK_STAT_INFO_INTERVAL)
        {/*{{{*/
            pthread_mutex_lock(&stat_info._svr_.lock); //lock
            struct _svr_info *svr;
            int i;
            for(i=0; i<SVR_MAX_NUM; ++i)
            {
                svr=&stat_info._svr_.svrs[i];
                if(svr->pid != -1)
                {
                    if(kill(svr->pid, 0) == -1)
                    {
                        errlog("%s: svr(%s) gone",
                                __func__, svr->name);
                        _free_svcs(svr);
                    }
                }
            }
            pthread_mutex_unlock(&stat_info._svr_.lock); //unlock

            last_stat=now;
        }/*}}}*/
    }
    errlog("%s: listener shutdown", __func__);
    DEEP_SLEEP;
}/*}}}*/

int listener_start(THRD_MSGQ *msgqs)
{/*{{{*/
    char buf[200];
    int res;

    // operate_info
    pthread_mutex_init(&operate_info._listener_.lock, NULL);
    operate_info._listener_.quit=IS_FALSE;
    operate_info._listener_.state=THRD_RUNNING;

    res=ep_create(&listener_ctl.ep);
    if(res)
    {
        sprintf(buf, "agent: listener(ep_create) fail, errcode:%d", res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }

    SOCKET sock;

    // server
    sock=sock_listen(cnfg_info.server_port);
    if(sock == INVALID_SOCKET)
    {
        sprintf(buf, "agent: open server port fail");
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    res=ep_slot_register(listener_ctl.ep, sock, IS_TRUE,
            &listener_ctl.server_slot);
    if(res)
    {
        sprintf(buf,
                "agent: listener(ep_slot_register) server port fail, "
                "errcode:%d",
                res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    sprintf(buf, "agent: open server port(%d)", cnfg_info.server_port);
    fprintf(stderr, "%s\n", buf); errlog(buf);

    // register
    char addr[LONG_FILE_NM_SIZE+1];
    sprintf(addr, "seagent_register_%d", cnfg_info.ipckey);
    sock=sock_local_listen(addr);
    if(sock == INVALID_SOCKET)
    {
        sprintf(buf, "agent: open register port(%s) fail", addr);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    res=ep_slot_register(listener_ctl.ep, sock, IS_TRUE,
            &listener_ctl.register_slot);
    if(res)
    {
        sprintf(buf,
                "agent: listener(ep_slot_register) register port fail, "
                "errcode:%d",
                res);
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }
    sprintf(buf, "agent: open register unix domain(%s)", addr);
    fprintf(stderr, "%s\n", buf); errlog(buf);

    ALLOC_MULTI_DOERR(listener_ctl.msg, char, SOCK_MSG_SIZE,
            sprintf(buf, "agent: listener alloc msg fail");
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );

    /*--- thread listener ---*/
    if(thrd_create((THRD_FUNC_T)_listener, (void *)NULL, THRD_DETACH,
                &listener_ctl.thrd_id))
    {
        sprintf(buf, "agent: listener thrd create fail");
        fprintf(stderr, "%s\n", buf); errlog(buf);
        return(-1);
    }

    sprintf(buf, "agent: listener create succeed");
    fprintf(stderr, "%s\n", buf); errlog(buf);
    return(0);
}/*}}}*/
